package com.lei.study.learn;

import com.lei.study.utils.JsonUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 * 原生的zk客户端实现的订阅发布场景
 * <p>
 * 1. 配置中心负责保存配置信息
 * 2. 客户端监听配置信息并打印出来，同时重新注册监听事件
 * </p>
 *
 * @author <a href="mailto:17674030991@163.com">伍磊</a>
 */
public class OriginalZookeeperPublishAndSubscribe {

    public static void main(String[] args) throws Exception {
        String configPath = "/configServer";
        String zkServer = "localhost:2181";

        // 1. 创建server
        ConfigServer configServer = new ConfigServer(configPath, zkServer);
        Map<String, String> map = new HashMap<>();
        map.put("name", "伍磊");
        configServer.publish(map);

        // 2. 创建client
        ConfigClient configClient = new ConfigClient(configPath, zkServer);
        configClient.sub();
        // new ConfigClient(configPath, zkServer);

        ConfigClient configClient1 = new ConfigClient(configPath, zkServer);
        configClient1.sub();


        TimeUnit.SECONDS.sleep(1);

        map.put("name", "伍磊1");
        map.put("age", "26");
        configServer.publish(map);

        System.in.read();
    }
}

/**
 * 配置中心
 */
class ConfigServer {

    private String configPath;

    private String zkServer;

    private ZooKeeper zooKeeper;

    public ConfigServer(String configPath, String zkServer) throws IOException {
        this.configPath = configPath;
        this.zkServer = zkServer;
        start();
    }

    private void start() throws IOException {
        zooKeeper = new ZooKeeper(zkServer, 30000, event -> System.out.println("连接事件: " + event));
    }

    /**
     * 更新配置信息
     *
     * @param infoMap
     */
    public void publish(Map<String, String> infoMap) throws InterruptedException, KeeperException {
        Stat exists = zooKeeper.exists(configPath, null);
        if (null == exists) {
            // 创建节点
            zooKeeper.create(configPath
                    , JsonUtils.serialize(infoMap)
                    , ZooDefs.Ids.OPEN_ACL_UNSAFE
                    , CreateMode.PERSISTENT);
            return;
        }

        // 写入节点数据
        int version = exists.getVersion();
        zooKeeper.setData(configPath, JsonUtils.serialize(infoMap), version);
    }
}

/**
 * 配置中心的客户端
 * 获取配置信息
 */
class ConfigClient {

    private String configPath;

    private String zkServer;

    private ZooKeeper zooKeeper;


    public ConfigClient(String configPath, String zkServer) throws IOException, InterruptedException, KeeperException {
        this.configPath = configPath;
        this.zkServer = zkServer;
        start();
    }

    private void start() throws IOException, InterruptedException, KeeperException {
        // 1. 启动
        zooKeeper = new ZooKeeper(zkServer, 30000, event -> {

        });
    }

    public void sub() throws InterruptedException, KeeperException {
        // 2. 监听事件
        Watcher watcher = new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                Event.EventType type = event.getType();
                if (type == Event.EventType.NodeDataChanged && configPath.equals(event.getPath())) {
                    // 获取数据并重新使用此监听器
                    byte[] data;
                    try {
                        data = zooKeeper.getData(configPath, this, null);
                    } catch (KeeperException | InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    System.out.println("数据发生变更: " + JsonUtils.deserialize(data, Map.class));
                }
            }
        };

        // 3. 注册监听事件
        byte[] data = zooKeeper.getData(configPath, watcher, null);
        System.out.println("原始数据: " + JsonUtils.deserialize(data, Map.class));
    }
}